package cn.itcast.streaming.flatmap;

import cn.itcast.streaming.entity.OnlineDataObj;
import cn.itcast.streaming.entity.VehicleInfoModel;
import cn.itcast.streaming.utils.DateUtil;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;

/**
 * Author itcast
 * Date 2021/9/27 17:42
 * 将窗口内的告警的车辆数据 onlineDataObj 和 数据库中的静态数据 车型、车系、销售日期等静态数据
 * 拉宽生成一个结果
 * 开发步骤：
 * 1.flatMap1
 * 2.flatMap2 将配置静态车辆的配置数据通过 vin 获取出来
 * 并与车辆数据拉宽
 */
public class VehicleInfoMapMysqlFunction extends RichCoFlatMapFunction<OnlineDataObj, HashMap<String, VehicleInfoModel>, OnlineDataObj> {
    private static final Logger logger = LoggerFactory.getLogger(VehicleInfoMapMysqlFunction.class);

    private static HashMap<String, VehicleInfoModel> vehicleInfoModelConfig = null;

    @Override
    public void flatMap1(OnlineDataObj value, Collector<OnlineDataObj> out) throws Exception {
        //1.1.通过 vin 获取到车辆基础信息
        String vin = value.getVin();
        VehicleInfoModel vehicleInfoModel = vehicleInfoModelConfig.getOrDefault(vin, null);

        //1.2.如果车辆基础信息不为空
        if (vehicleInfoModel != null) {
            //1.2.1.将车系seriesName，车型modelName,年限LiveTime,销售日期saleDate，车辆类型carType封装到onlineDataObj对象中
            value.setModelName(vehicleInfoModel.getModelName());
            value.setSeriesName(vehicleInfoModel.getSeriesName());
            value.setLiveTime(
                    (System.currentTimeMillis() - DateUtil.convertStringToDateTime(vehicleInfoModel.getSalesDate())
                            .getTime()) / 1000 / 3600 / 24 / 30 + ""
            );
            value.setSalesDate(vehicleInfoModel.getSalesDate());
            value.setCarType(vehicleInfoModel.getCarType());

            //1.2.2.将onlineDataObj收集返回
            out.collect(value);
            //1.3.打印输出，基础信息不存在
        }else{
            logger.error("当前上报车辆的位置数据不存在");
        }

    }

    @Override
    public void flatMap2(HashMap<String, VehicleInfoModel> value, Collector<OnlineDataObj> out) throws Exception {
        vehicleInfoModelConfig = value;
    }
}
